package com.lemon.cloud.rocketmq.nacos;

import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.nacos.api.naming.listener.Event;
import com.alibaba.nacos.api.naming.listener.NamingEvent;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.lemon.cloud.config.NacosWatchEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
import org.apache.rocketmq.common.ServiceState;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.springframework.boot.availability.ApplicationAvailability;
import org.springframework.boot.availability.LivenessState;
import org.springframework.cloud.commons.util.InetUtils;
import org.springframework.core.env.Environment;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j
public class RocketEvent implements NacosWatchEvent {

    private final ReentrantLock lock = new ReentrantLock();
    private final AtomicBoolean isProcessing = new AtomicBoolean(false);
    private static final int MAX_RETRY_ATTEMPTS = 3;
    private static final long SHUTDOWN_TIMEOUT_MILLIS = 30000; // 延长关闭超时时间

    @Override
    public String getKey() {
        return "rocketmq.name-server";
    }

    @Override
    public boolean accept(String value) {
        return StrUtil.isNotEmpty(value);
    }

    @Override
    public void doEvent(Event event) {
        if (isRunning()) {
            log.info("Application is not in CORRECT state, skipping event processing");
            return;
        }

        if (isProcessing.get()) {
            log.warn("Event processing is already in progress, skipping");
            return;
        }

        Environment environment = SpringUtil.getBean(Environment.class);
        String appName = environment.getProperty(NacosWatchEvent.APP_NAME);
        String port = environment.getProperty(NacosWatchEvent.APP_PORT);
        if (StrUtil.isEmpty(appName) || StrUtil.isEmpty(port)) {
            log.error("Configuration error: APP_NAME [{}] or APP_PORT [{}] is empty", appName, port);
            return;
        }

        if (event instanceof NamingEvent) {
            processNamingEvent((NamingEvent) event, appName, port);
        }
    }

    private void processNamingEvent(NamingEvent event, String appName, String port) {
        lock.lock();
        try {
            if (isProcessing.compareAndSet(false, true)) {
                List<Instance> instances = event.getInstances();
                String ip = getIp();
                boolean isOnline = instances.stream().anyMatch(i -> isCurrInstance(i, ip, port) && isCurrService(i, appName));
                log.info("Processing NamingEvent for instance [{}:{}], service [{}], online: {}", ip, port, appName, isOnline);

                Map<String, DefaultRocketMQListenerContainer> containerMap = new ConcurrentHashMap<>(
                        SpringUtil.getBeansOfType(DefaultRocketMQListenerContainer.class));

                if (!isOnline) {
                    shutdownConsumers(containerMap, ip, port);
                } else {
                    startConsumers(containerMap, ip, port);
                }
            }
        } finally {
            isProcessing.set(false);
            lock.unlock();
        }
    }

    private void shutdownConsumers(Map<String, DefaultRocketMQListenerContainer> containerMap, String ip, String port) {
        containerMap.forEach((name, container) -> {
            if (container.isRunning()) {
                int attempt = 0;
                while (attempt < MAX_RETRY_ATTEMPTS) {
                    try {
                        DefaultMQPushConsumer consumer = container.getConsumer();
                        consumer.setAwaitTerminationMillisWhenShutdown(SHUTDOWN_TIMEOUT_MILLIS);
                        container.stop();
                        log.info("Successfully stopped RocketMQ consumer [{}] for instance [{}:{}]", name, ip, port);
                        break;
                    } catch (Exception e) {
                        attempt++;
                        log.warn("Failed to stop RocketMQ consumer [{}] (attempt {}/{}), error: {}", name, attempt, MAX_RETRY_ATTEMPTS, e.getMessage());
                        if (attempt == MAX_RETRY_ATTEMPTS) {
                            log.error("Max retries reached for stopping consumer [{}], giving up", name);
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            log.error("Interrupted while retrying to stop consumer [{}]", name);
                        }
                    }
                }
            }
        });
        log.warn("Instance [{}:{}] RocketMQ consumers are down", ip, port);
    }

    private void startConsumers(Map<String, DefaultRocketMQListenerContainer> containerMap, String ip, String port) {
        containerMap.forEach((name, container) -> {
            if (!container.isRunning()) {
                int attempt = 0;
                while (attempt < MAX_RETRY_ATTEMPTS) {
                    try {
                        DefaultMQPushConsumer consumer = container.getConsumer();
                        DefaultMQPushConsumerImpl defaultMQPushConsumerImpl = consumer.getDefaultMQPushConsumerImpl();
                        defaultMQPushConsumerImpl.setServiceState(ServiceState.CREATE_JUST);
                        container.start();
                        log.info("Successfully started RocketMQ consumer [{}] for instance [{}:{}], group: {}",
                                name, ip, port, consumer.getConsumerGroup());
                        break;
                    } catch (Exception e) {
                        attempt++;
                        log.warn("Failed to start RocketMQ consumer [{}] (attempt {}/{}), error: {}", name, attempt, MAX_RETRY_ATTEMPTS, e.getMessage());
                        if (attempt == MAX_RETRY_ATTEMPTS) {
                            log.error("Max retries reached for starting consumer [{}], giving up", name);
                        }
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException ie) {
                            Thread.currentThread().interrupt();
                            log.error("Interrupted while retrying to start consumer [{}]", name);
                        }
                    }
                }
            }
        });
    }

    private boolean isRunning() {
        ApplicationAvailability availability = SpringUtil.getBean(ApplicationAvailability.class);
        boolean isCorrectState = !LivenessState.CORRECT.equals(availability.getLivenessState());
        log.debug("Application liveness state: {}, isRunning: {}", availability.getLivenessState(), isCorrectState);
        return isCorrectState;
    }

    private String getIp() {
        InetUtils inetUtils = SpringUtil.getBean(InetUtils.class);
        String ip = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
        log.debug("Retrieved local IP: {}", ip);
        return ip;
    }

    private boolean isCurrInstance(Instance i, String ip, String port) {
        boolean isMatch = ip.equals(i.getIp()) && port.equals(String.valueOf(i.getPort()));
        log.debug("Checking instance [{}:{}], match: {}", i.getIp(), i.getPort(), isMatch);
        return isMatch;
    }

    private boolean isCurrService(Instance i, String appName) {
        boolean isMatch = StrUtil.isNotEmpty(i.getServiceName()) && i.getServiceName().contains(StrUtil.AT + appName);
        log.debug("Checking service [{}], expected [{}], match: {}", i.getServiceName(), appName, isMatch);
        return isMatch;
    }
}